#include <rtc_base/logging.h>
#include <unistd.h>
#include <fcntl.h>
#include "stream/rtc_stream_manager.h"
#include "stream/push_stream.h"
#include "stream/pull_stream.h"
#include "base/conf.h"
#include "server/rtc_worker.h"
#include "server/rtc_server.h"
extern xrtc::GeneralConf *gconf;
extern xrtc::RtcServer *g_rtc_server;
namespace xrtc
{
    namespace
    {

        void RtcStreamManagerRecvNotify(EventLoop *, IOWatcher *, int fd, int /*event*/, void *data)
        {
            int type;
            if (read(fd, &type, sizeof(type)) != sizeof(int))
            {
                RTC_LOG(LS_WARNING) << "read from rtc stream manager pipe failed!";
                return;
            }
            RtcStreamManager *mgr = (RtcStreamManager *)data;
            mgr->processNotify(type);
        }

    } // namespace

    RtcStreamManager::RtcStreamManager(EventLoop *el) : el_(el), allocator_(new PortAllocator())
    {
        allocator_->SetPortRange(gconf->ice_min_port, gconf->ice_max_port);
        for (int i = 0; i < g_rtc_server->worker_num(); ++i)
        {
            if (i == GetWorkerId())
            {
                continue;
            }
            DeliverPacketQueue *q = new DeliverPacketQueue();
            queue_map_.emplace(i, q);
        }
    }

    RtcStreamManager::~RtcStreamManager()
    {
        PullStreamMap::iterator pit = multi_pull_streams_.begin();
        while (pit != multi_pull_streams_.end())
        {
            if (pit->second)
            {
                pit->second->clear();
                delete pit->second;
                pit->second = nullptr;
            }

            ++pit;
        }
        multi_pull_streams_.clear();
        for (auto it : queue_map_)
        {
            if (it.second)
            {
                delete it.second;
            }
        }
        queue_map_.clear();

        if (pipe_watcher_)
        {
            el_->DeleteIOEvent(pipe_watcher_);
            pipe_watcher_ = nullptr;
            close(notify_recv_fd_);
            close(notify_send_fd_);
        }
    }
    int RtcStreamManager::Init()
    {
        int fds[2];
        if (pipe(fds))
        {
            RTC_LOG(LS_WARNING) << "create rtc stream manager pipe failed!";
            return -1;
        }
        notify_recv_fd_ = fds[0];
        notify_send_fd_ = fds[1];
        fcntl(notify_send_fd_, F_SETFL, fcntl(notify_send_fd_, F_GETFL) | O_NONBLOCK);
        pipe_watcher_ = el_->CreateIOEvent(RtcStreamManagerRecvNotify, this);
        el_->StartIOEvent(pipe_watcher_, notify_recv_fd_, EventLoop::READ);

        return 0;
    }
    void RtcStreamManager::processNotify(int type)
    {
        if (type >= 0 && type <= CM_NEW_DELIVER_PACKET_END)
        {
            // 新数据包到达的消息
            return ProcessNewDeliverPacket(type);
        }
        std::shared_ptr<ComMsg> msg;
        { //{} 保证锁作用范围在{} 内

            std::unique_lock<std::mutex> lock(mq_mutex_);
            mq_.Consume(&msg);
        }
        if (!msg)
        {
            return;
        }
        switch (msg->msg_type)
        {
        case CM_NEW_PUSH_STREAM:
            ProcessNewPushStream(msg);
            break;
        case CM_REMOVE_PUSH_STREAM:
            ProcessRemovePushStream(msg);
            break;
        case CM_NEW_PULL_STREAM:
            ProcessNewPullStream(msg);
            break;
        case CM_REMOVE_PULL_STREAM:
            ProcessRemovePullStream(msg);
            break;
        default:
            RTC_LOG(LS_WARNING) << "unknow msg type: " << msg->msg_type;
            break;
        }
    }
    void RtcStreamManager::ProcessRemovePushStream(std::shared_ptr<ComMsg> msg)
    {
        RTC_LOG(LS_INFO) << ">>>>>>>>>>>>>>>>>>>>>>>>>>>>>process remove push stream,sender worker_id:" << msg->worker_id
                         << ",cur worker_id:" << GetWorkerId();
        auto it = push_stream_sources_.find(msg->stream_name);
        if (it != push_stream_sources_.end())
        {
            push_stream_sources_.erase(it);
        }
    }
    void RtcStreamManager::ProcessNewPushStream(std::shared_ptr<ComMsg> msg)
    {
        RTC_LOG(LS_INFO) << "************process new push stream,sender worker_id:" << msg->worker_id
                         << ",cur worker_id:" << GetWorkerId();
        std::shared_ptr<PushStreamSource> source = std::make_shared<PushStreamSource>(msg->uid, msg->stream_name, msg->audio_source, msg->video_source);
        push_stream_sources_[msg->stream_name] = source;
    }
    void RtcStreamManager::ProcessNewPullStream(std::shared_ptr<ComMsg> msg)
    {
        RTC_LOG(LS_INFO) << "************process new pull stream,sender worker_id:" << msg->worker_id
                         << ",cur worker_id:" << GetWorkerId();
        // 源流所在的worker 线程条用该方法
        auto it = pull_workers_.find(msg->stream_name);
        if (it == pull_workers_.end())
        {
            // 第一个线程来注册
            std::set<int> worker_ids;
            worker_ids.insert(msg->worker_id);
            pull_workers_[msg->stream_name] = worker_ids;
        }
        else
        {
            it->second.insert(msg->worker_id);
        }
    }
    int RtcStreamManager::Notify(int type)
    {
        int written = write(notify_send_fd_, &type, sizeof(type));
        if (written != sizeof(int))
        {
            RTC_LOG(LS_WARNING) << "write rtc stream manager pip failed!";
            return -1;
        }
        return 0;
    }
    int RtcStreamManager::NotifyMsg(std::shared_ptr<ComMsg> msg)
    {
        std::unique_lock<std::mutex> lock(mq_mutex_);
        mq_.Produce(msg);
        return Notify(msg->msg_type);
    }

    PushStream *RtcStreamManager::FindPushStream(const std::string &stream_name)
    {

        auto iter = push_streams_.find(stream_name);
        if (iter != push_streams_.end())
        {
            return iter->second;
        }
        return nullptr;
    }
    PullStream *RtcStreamManager::FindPullStream(const std::string &stream_name)
    {

        auto iter = pull_streams_.find(stream_name);
        if (iter != pull_streams_.end())
        {
            return iter->second;
        }
        return nullptr;
    }
    int RtcStreamManager::SetAnswer(uint64_t uid,
                                    const std::string &stream_name,
                                    std::string &answer,
                                    std::string &stream_type,
                                    const std::string &mode,
                                    uint32_t log_id)
    {
        // 判断是拉流还是推流
        if ("push" == stream_type)
        {
            PushStream *push_stream = FindPushStream(stream_name);
            if (!push_stream)
            {
                RTC_LOG(LS_WARNING) << "push stream not found ,uid : " << uid
                                    << ", stream_name: " << stream_name
                                    << ", log_id: " << log_id;
                return -1;
            }
            if (uid != push_stream->uid)
            {

                RTC_LOG(LS_WARNING) << "uid invalid,uid : " << uid
                                    << ", stream_name: " << stream_name
                                    << ", log_id: " << log_id;
                return -1;
            }
            push_stream->SetRemoteSdp(answer);
        }
        else if ("pull" == stream_type)
        {
            PullStream *pull_stream = nullptr;
            if ("transparent" == mode)
            {
                pull_stream = FindPullStream(stream_name);
            }
            else
            {
                pull_stream = FindPullStreamM(uid, stream_name);
            }

            if (!pull_stream)
            {
                RTC_LOG(LS_WARNING) << "pull stream not found ,uid : " << uid
                                    << ", stream_name: " << stream_name
                                    << ", log_id: " << log_id;
                return -1;
            }
            if (uid != pull_stream->uid)
            {

                RTC_LOG(LS_WARNING) << "uid invalid,uid : " << uid
                                    << ", stream_name: " << stream_name
                                    << ", log_id: " << log_id;
                return -1;
            }
            pull_stream->SetRemoteSdp(answer);
        }
        return 0;
    }

    int RtcStreamManager::CreatePushStream(uint64_t uid,
                                           const std::string &stream_name,
                                           bool audio,
                                           bool video,
                                           bool is_dtls,
                                           bool is_pli,
                                           const std::string &mode,
                                           uint32_t log_id,
                                           rtc::RTCCertificate *certificate,
                                           std::string &offer)
    {
        // 检查stream_name 是否已经存在
        PushStream *stream = FindPushStream(stream_name);
        if (stream)
        {
            push_streams_.erase(stream_name);
            delete stream;
        }
        stream = new PushStream(el_, allocator_.get(), uid, stream_name, audio, video, log_id);
        stream->RegisterListener(this);
        stream->set_mode(mode);
        stream->set_pli(is_pli);
        if (is_dtls)
        {
            stream->Start(certificate);
        }
        else
        {
            stream->Start(nullptr);
        }

        offer = stream->CreateOffer();
        // 添加到map
        push_streams_[stream_name] = stream;
        return 0;
    }
    std::shared_ptr<PushStreamSource> RtcStreamManager::FindPushStreamSource(
        const std::string &stream_name)
    {
        auto it = push_stream_sources_.find(stream_name);
        if (it != push_stream_sources_.end())
        {
            return it->second;
        }
        return nullptr;
    }
    int RtcStreamManager::CreatePullStream(uint64_t uid,
                                           const std::string &stream_name,
                                           bool audio,
                                           bool video,
                                           bool is_dtls,
                                           const std::string &mode,
                                           uint32_t log_id,
                                           rtc::RTCCertificate *certificate,
                                           std::string &offer)
    {
        std::vector<StreamParams> audio_source;
        std::vector<StreamParams> video_source;
        // 检查是否存在推流stream_name 是否已经存在，才能拉流
        PushStream *push_stream = FindPushStream(stream_name);
        if (!push_stream)
        {
            RTC_LOG(LS_WARNING) << "push stream not found ,uid : " << uid
                                << ", stream_name: " << stream_name
                                << ", log_id: " << log_id;
            if ("transparent" == mode)
            {
                return -1;
            }
            else
            {

                auto push_stream_source = FindPushStreamSource(stream_name);
                if (!push_stream_source)
                {
                    RTC_LOG(LS_WARNING) << " push stream source not found ,uid : " << uid
                                        << ", stream_name: " << stream_name
                                        << ", log_id: " << log_id;
                    return -1;
                }
                audio_source = push_stream_source->audio_source;
                video_source = push_stream_source->video_source;
            }
        }

        if ("transparent" == mode)
        {
            RemovePullStream(uid, stream_name);
        }
        else
        {
            RemovePullStreamM(uid, stream_name);
        }
        if (push_stream)
        {
            push_stream->GetAudioSource(audio_source);
            push_stream->GetVideoSource(video_source);
        }

        PullStream *stream = new PullStream(el_, allocator_.get(), uid, stream_name, audio, video, log_id);
        stream->RegisterListener(this);
        stream->set_mode(mode);
        stream->AddAudioSource(audio_source);
        stream->AddVideoSource(video_source);
        if (is_dtls)
        {
            stream->Start(certificate);
        }
        else
        {
            stream->Start(nullptr);
        }

        offer = stream->CreateOffer();
        // 添加到map
        if (stream->IsTransparent())
        {
            pull_streams_[stream_name] = stream;
        }
        else
        {
            // 直播模式
            AddPullStreamM(stream);
        }
        return 0;
    }
    void RtcStreamManager::RemovePushStream(RtcStream *stream)
    {
        if (!stream)
        {
            return;
        }
        RemovePushStream(stream->GetUid(), stream->GetStreamName());
    }
    void RtcStreamManager::RemovePushStream(uint64_t uid, const std::string &stream_name)
    {
        PushStream *push_stream = FindPushStream(stream_name);
        if (push_stream && uid == push_stream->GetUid())
        {
            // 通知其他的worker线程清理流的注册信息
            NotifyRemovePushStream(push_stream);
            push_streams_.erase(stream_name);
            delete push_stream;
        }
    }
    void RtcStreamManager::AddPullStreamM(PullStream *stream)
    {
        if (!stream)
        {
            return;
        }
        UserStreamMap *umap = nullptr;
        PullStreamMap::iterator pit = multi_pull_streams_.find(stream->GetStreamName());
        if (pit != multi_pull_streams_.end())
        {
            umap = pit->second;
        }
        else
        {
            // 第一次被拉流
            umap = new UserStreamMap();
            multi_pull_streams_[stream->GetStreamName()] = umap;
            if (umap)
            {
                umap->insert(std::pair<uint64_t, PullStream *>(stream->GetUid(), stream));
            }
        }
    }
    void RtcStreamManager::RemovePullStreamM(RtcStream *stream)
    {
        if (!stream)
        {
            return;
        }
        RemovePullStreamM(stream->GetUid(), stream->GetStreamName());
    }
    void RtcStreamManager::RemovePullStreamM(uint64_t uid, const std::string &stream_name)
    {
        PullStreamMap::iterator pit = multi_pull_streams_.find(stream_name);
        if (pit == multi_pull_streams_.end())
        {
            return;
        }
        UserStreamMap *umap = pit->second;
        if (!umap)
        {
            return;
        }
        UserStreamMap::iterator uit = umap->find(uid);
        if (uit != umap->end())
        {
            PullStream *pull_stream = uit->second;
            umap->erase(uid);
            delete pull_stream;
        }
        if (umap->empty())
        {
            multi_pull_streams_.erase(pit);
            delete umap;
            // 表示当前的woker 已经没有用户拉去这个流
            NotifyRemovePullStream(stream_name);
        }
    }
    void RtcStreamManager::OnConnectionState(RtcStream *stream, PeerConnectionState state)
    {
        if (state == PeerConnectionState::kFailed)
        {
            if (stream->StreamType() == RtcStreamType::kPush)
            {
                RemovePushStream(stream);
            }
            else if (stream->StreamType() == RtcStreamType::kPull)
            {
                if (stream->IsTransparent())
                {
                    RemovePullStream(stream);
                }
                else
                {
                    RemovePullStreamM(stream);
                }
            }
        }
        else if (state == PeerConnectionState::kConnected)
        {
            if (stream->StreamType() == RtcStreamType::kPush)
            {
                PushStream *push_stream = dynamic_cast<PushStream *>(stream);
                NotifyNewPushStream(push_stream);
            }
            else if (stream->StreamType() == RtcStreamType::kPull)
            {
                // 源流是否在当前的 worker 线程
                if (!FindPushStream(stream->GetStreamName()))
                {
                    // 源流不在当前的 worker 线程，向源流所在的woker 发起订阅请求
                    PullStream *pull_stream = dynamic_cast<PullStream *>(stream);
                    NotifyNewPullStream(pull_stream);
                }
            }
        }
    }
    int RtcStreamManager::GetWorkerId()
    {
        RtcWorker *worker = (RtcWorker *)(el_->Owner());
        return worker->worker_id();
    }

    void RtcStreamManager::NotifyRemovePullStream(const std::string &stream_name)
    {

        // 如果stream_name 就在当前的线程，不需要通知
        auto push_stream = FindPushStream(stream_name);
        if (push_stream)
        {
            return;
        }
        std::shared_ptr<ComMsg> msg = std::make_shared<ComMsg>();
        msg->msg_type = CM_REMOVE_PULL_STREAM;
        msg->worker_id = GetWorkerId();
        msg->stream_name = stream_name;
        // 向源流所在的线程发起注册通知
        RtcWorker *worker = g_rtc_server->GetWorker(msg->stream_name);
        if (worker)
        {
            worker->rtc_stream_mgr()->NotifyMsg(msg);
        }
    }

    void RtcStreamManager::NotifyRemovePushStream(PushStream *push_stream)
    {
        std::shared_ptr<ComMsg> msg = std::make_shared<ComMsg>();
        msg->msg_type = CM_REMOVE_PUSH_STREAM;
        msg->worker_id = GetWorkerId();
        msg->uid = push_stream->GetUid();
        msg->stream_name = push_stream->GetStreamName();
        // push_stream->GetAudioSource(msg->audio_source);
        // push_stream->GetVideoSource(msg->video_source);
        for (auto worker : g_rtc_server->workers())
        {
            // 向其他的worker 线程发送清除注册消息
            if (msg->worker_id != worker->worker_id())
            {
                worker->rtc_stream_mgr()->NotifyMsg(msg);
            }
        }
    }
    void RtcStreamManager::NotifyNewPushStream(PushStream *push_stream)
    {
        std::shared_ptr<ComMsg> msg = std::make_shared<ComMsg>();
        msg->msg_type = CM_NEW_PUSH_STREAM;
        msg->worker_id = GetWorkerId();
        msg->uid = push_stream->GetUid();
        msg->stream_name = push_stream->GetStreamName();
        push_stream->GetAudioSource(msg->audio_source);
        push_stream->GetVideoSource(msg->video_source);
        for (auto worker : g_rtc_server->workers())
        {
            // 向其他的worker 线程发送新流注册消息
            if (msg->worker_id != worker->worker_id())
            {
                worker->rtc_stream_mgr()->NotifyMsg(msg);
            }
        }
    }
    void RtcStreamManager::NotifyNewPullStream(PullStream *pull_stream)
    {
        // 检测是否该 worker第一个拉流用户，如果不是的话，不需要重复注册
        auto pit = multi_pull_streams_.find(pull_stream->GetStreamName());
        if (pit != multi_pull_streams_.end())
        {
            UserStreamMap *umap = pit->second;
            if (umap && umap->size() > 1)
            {
                return;
            }
        }

        std::shared_ptr<ComMsg> msg = std::make_shared<ComMsg>();
        msg->msg_type = CM_NEW_PULL_STREAM;
        msg->worker_id = GetWorkerId();
        msg->uid = pull_stream->GetUid();
        msg->stream_name = pull_stream->GetStreamName();
        // 向源流所在的线程发起注册通知
        RtcWorker *worker = g_rtc_server->GetWorker(msg->stream_name);
        if (worker)
        {
            worker->rtc_stream_mgr()->NotifyMsg(msg);
        }
    }
    int RtcStreamManager::StopPush(uint64_t uid, const std::string &stream_name)
    {
        RemovePushStream(uid, stream_name);
        return 0;
    }
    int RtcStreamManager::StopPull(uint64_t uid, const std::string &stream_name, const std::string &mode)
    {
        if ("transparent" == mode)
        {
            RemovePullStream(uid, stream_name);
        }
        else
        {
            RemovePullStreamM(uid, stream_name);
        }
        return 0;
    }

    void RtcStreamManager::RemovePullStream(uint64_t uid, const std::string &stream_name)
    {
        PullStream *pull_stream = FindPullStream(stream_name);
        if (pull_stream && uid == pull_stream->GetUid())
        {
            pull_streams_.erase(stream_name);
            delete pull_stream;
        }
    }
    void RtcStreamManager::RemovePullStream(RtcStream *stream)
    {
        if (!stream)
        {
            return;
        }
        RemovePullStream(stream->GetUid(), stream->GetStreamName());
    }
    // transport
    void RtcStreamManager::OnRtpPacketReceived(RtcStream *stream, const char *data, size_t len)
    {
        if (RtcStreamType::kPush == stream->StreamType())
        {
            if (stream->IsTransparent()) // 这个可以不加因为之后透传模式才能进入这个函数，live 模式走的是另一个函数
            {
                PullStream *pull_stream = FindPullStream(stream->GetStreamName());
                if (pull_stream)
                {
                    pull_stream->SendRtp(data, len);
                }
            }
        }
    }
    void RtcStreamManager::OnRtcpPacketReceived(RtcStream *stream, const char *data, size_t len)
    {
        if (stream->IsTransparent()) //  这个判断模式也可以去掉，因为live模式下不走这个函数
        {
            if (RtcStreamType::kPush == stream->StreamType())
            {
                PullStream *pull_stream = FindPullStream(stream->GetStreamName());
                if (pull_stream)
                {
                    pull_stream->SendRtcp(data, len);
                }
            }
            else if (RtcStreamType::kPull == stream->StreamType())
            {
                PushStream *push_stream = FindPushStream(stream->GetStreamName());
                if (push_stream)
                {
                    push_stream->SendRtcp(data, len);
                }
            }
        }
    }
    void RtcStreamManager::OnStreamException(RtcStream *stream)
    {
        if (RtcStreamType::kPush == stream->StreamType())
        {
            RemovePushStream(stream);
        }
        else if (RtcStreamType::kPull == stream->StreamType())
        {
            if (stream->IsTransparent())
            {
                RemovePullStream(stream);
            }
            else
            {
                RemovePullStreamM(stream);
            }
        }
    }
    PullStream *RtcStreamManager::FindPullStreamM(uint64_t uid, const std::string &stream_name)
    {

        PullStreamMap::iterator pit = multi_pull_streams_.find(stream_name);
        if (pit == multi_pull_streams_.end())
        {
            return nullptr;
        }
        UserStreamMap *umap = pit->second;
        if (!umap)
        {
            return nullptr;
        }
        UserStreamMap::iterator uit = umap->find(uid);
        if (uit != umap->end())
        {
            return uit->second;
        }
        return nullptr;
    }
    void RtcStreamManager::OnRtpPacket(RtcStream *stream, webrtc::MediaType media_type,
                                       std::shared_ptr<RtcPacket> packet)
    {
        // RTC_LOG(LS_WARNING) << "====seq:" << packet.SequenceNumber();
        // 转发给拉流端
        if (RtcStreamType::kPush == stream->StreamType())
        {
            // 通知其他的注册订阅该流的线程，有新的数据包到达
            DeliverNewPacket(stream->GetStreamName(), media_type, packet);

            // 向本线程管理的pull_stream 转发数据
            SendPacket(stream->GetStreamName(), media_type, packet);
        }
    }
    void RtcStreamManager::OnSrInfo(RtcStream *stream, webrtc::MediaType media_type, uint32_t rtp_timestamp,
                                    webrtc::NtpTime ntp)
    {
        if (stream->IsTransparent())
        {
            return;
        }
        // RTC_LOG(LS_WARNING) << "====on sr info,media_Type: " << (int)media_type
        //                     << ",rtp_timestamp: " << rtp_timestamp
        //                     << ",ntp: " << ntp.ToMs();
        if (RtcStreamType::kPush == stream->StreamType())
        {
            PullStreamMap::iterator pit = multi_pull_streams_.find(stream->GetStreamName());
            if (pit == multi_pull_streams_.end())
            {
                return;
            }
            UserStreamMap *umap = pit->second;
            if (!umap)
            {
                return;
            }
            UserStreamMap::iterator uit = umap->begin();
            for (; uit != umap->end(); ++uit)
            {
                PullStream *pull_stream = uit->second;
                if (pull_stream)
                {
                    pull_stream->SetSrInfo(media_type, rtp_timestamp, ntp);
                }
            }
        }
    }

    void RtcStreamManager::OnNackReceived(RtcStream *stream, webrtc::MediaType media_type, const std::vector<uint16_t> &nack_list)
    {
        // RTC_LOG(LS_WARNING) << "====nack_size:" << nack_list.size();
        if (stream->StreamType() != RtcStreamType::kPull)
        {
            return;
        }
        PushStream *push_stream = FindPushStream(stream->GetStreamName());
        if (!push_stream)
        {
            return;
        }
        PullStream *pull_stream = dynamic_cast<PullStream *>(stream);
        for (auto seq_num : nack_list)
        {
            auto packet = push_stream->FindVideoPacket(seq_num);
            if (packet)
            {
                pull_stream->SendPacket(media_type, packet, true);
            }
        }
    }
    void RtcStreamManager::DeliverNewPacket(const std::string &stream_name, webrtc::MediaType media_type, std::shared_ptr<RtcPacket> packet)
    {
        auto iter = pull_workers_.find(stream_name);
        if (iter == pull_workers_.end())
        {
            return;
        }
        if (iter->second.empty())
        {
            return;
        }
        auto deliver_packet = std::make_shared<DeliverPacket>();
        deliver_packet->stream_name = stream_name;
        deliver_packet->media_type = media_type;
        deliver_packet->packet = packet;

        for (int worker_id : iter->second)
        {
            RtcWorker *worker = g_rtc_server->GetWorkerById(worker_id);
            if (worker)
            {
                worker->rtc_stream_mgr()->NotifyNewPacket(GetWorkerId(), deliver_packet);
            }
        }
    }
    DeliverPacketQueue *RtcStreamManager::FindQueueByWorkerId(int worker_id)
    {

        auto iter = queue_map_.find(worker_id);
        if (iter != queue_map_.end())
        {
            return iter->second;
        }
        return nullptr;
    }
    void RtcStreamManager::NotifyNewPacket(int worker_id, std::shared_ptr<DeliverPacket> packet)
    {

        // 查询该worker_id 对应的queue
        DeliverPacketQueue *queue = FindQueueByWorkerId(worker_id);
        if (!queue)
        {
            return;
        }
        queue->Produce(packet);
        Notify(worker_id);
    }
    void RtcStreamManager::ProcessNewDeliverPacket(int worker_id)
    {

        DeliverPacketQueue *queue = FindQueueByWorkerId(worker_id);
        if (!queue)
        {
            return;
        }
        std::shared_ptr<DeliverPacket> packet;
        queue->Consume(&packet);
        if (!packet)
        {
            return;
        }

        SendPacket(packet->stream_name, packet->media_type, packet->packet);
    }
    void RtcStreamManager::SendPacket(const std::string &stream_name, webrtc::MediaType media_type, std::shared_ptr<RtcPacket> packet)
    {

        PullStreamMap::iterator pit = multi_pull_streams_.find(stream_name);
        if (pit == multi_pull_streams_.end())
        {
            return;
        }
        UserStreamMap *umap = pit->second;
        if (!umap)
        {
            return;
        }
        UserStreamMap::iterator uit = umap->begin();
        for (; uit != umap->end(); ++uit)
        {
            PullStream *pull_stream = uit->second;
            if (pull_stream)
            {
                //  pull_stream->SendRtp((const char *)packet.data(), packet.size());
                RTC_LOG(LS_WARNING) << "==== worker_id :" << GetWorkerId()
                                    << ",stream_name :" << stream_name
                                    << ",seq:" << packet->seq_num
                                    << ",uid :" << pull_stream->GetUid();
                pull_stream->SendPacket(media_type, packet);
            }
        }
    }
    void RtcStreamManager::ProcessRemovePullStream(std::shared_ptr<ComMsg> msg)
    {

        RTC_LOG(LS_INFO) << "************process remove pull stream,sender worker_id:" << msg->worker_id
                         << ",cur worker_id:" << GetWorkerId();
        // 源流所在的worker 线程条用该方法
        auto it = pull_workers_.find(msg->stream_name);
        if (it != pull_workers_.end())
        {
            it->second.erase(msg->worker_id);
        }
    }

} // namespace xrtc